package book.test03;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.log4j.BasicConfigurator;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;


/**
 * 分析获得每个城市，职位名称中包含“数据分析”的职位需求、公司名称﹐并按照需求数量倒序排列，返回结果包含城市，公司名称，需求数量。
 */
public class MRTest03 {

    public static class NameMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
        Text k = new Text();
        IntWritable v = new IntWritable(1);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String[] data = value.toString().split(",(?=(?:[^\"]*\"[^\"]*\")*[^\"]*$)", -1);
            //如果职位名称里面没有包含数据分析那么就舍去
            if (!(data[0].contains("数据分析"))) return;
            k.set(data[5] + "\t" + data[6]);
            context.write(k, v);
        }
    }

    public static class NameReduce extends Reducer<Text, IntWritable, Text, NullWritable> {
        Text k = new Text();

        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            int count = 0;
            for (IntWritable value : values) {
                count += value.get();
            }
            k.set(key.toString().split("\t")[0]+"\t"+
                    key.toString().split("\t")[1]+ "\t" +
                    count);
            context.write(k, NullWritable.get());
        }
    }

    public static class NameMapper02 extends Mapper<LongWritable, Text, JavaBean, NullWritable> {

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String data[] = value.toString().split("\t");
            JavaBean k = new JavaBean();
            k.setCity(data[0]);
            k.setCompanyName(data[1]);
            k.setCount(Integer.parseInt(data[2]));
            context.write(k, NullWritable.get());
        }
    }

    //    public static class NameReduce02 extends Reducer<JavaBean,NullWritable,JavaBean,NullWritable>{
//
//        @Override
//        protected void reduce(JavaBean key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
//            for (NullWritable value:values){
//                context.write(key,NullWritable.get());
//            }
//        }
//    }
    public static void main(String[] args) throws Exception {
        BasicConfigurator.configure();
        Configuration conf = new Configuration();
        conf.set("fs.default", "hdfs://192.168.0.155:9000/");

        Job job = Job.getInstance(conf);

        job.setJarByClass(MRTest03.class);
        job.setMapperClass(NameMapper.class);
        job.setReducerClass(NameReduce.class);


        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job, "/MR/test03/input");
        Path path = new Path("/MR/test03/out");
        if (path.getFileSystem(conf).exists(path)) {
            path.getFileSystem(conf).delete(path);
        }
        FileOutputFormat.setOutputPath(job, path);
        boolean b = job.waitForCompletion(true);
        if (b) {
            Configuration conf2 = new Configuration();
            Job job2 = Job.getInstance(conf2);
            conf2.set("fs.default", "hdfs://192.168.0.155:9000/");

            job2.setJarByClass(MRTest03.class);
            job2.setMapperClass(NameMapper02.class);

            job2.setMapOutputKeyClass(JavaBean.class);
            job2.setOutputValueClass(NullWritable.class);

            job2.setOutputKeyClass(JavaBean.class);
            job2.setOutputValueClass(NullWritable.class);

            FileInputFormat.setInputPaths(job2, "/MR/test03/out/part-r-00000");
            Path path2 = new Path("/MR/test03/out2");
            if (path2.getFileSystem(conf2).exists(path2)) {
                path2.getFileSystem(conf2).delete(path2);
            }
            FileOutputFormat.setOutputPath(job2, path2);
            System.exit(job2.waitForCompletion(true) ? 0 : 1);
        }
    }
}

